package demo.kafka;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class KafkaConsumerDemo {
    //    public static final String KFKCONTPC = "test";
    public static final String KFKCONTPC = "fdm_testuserinfo";
    public static final String KFKADD = "10.10.15.243:9092,10.10.15.224:9092,10.10.15.222:9092";

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        try {
            //            env.setStateBackend(new
            // RocksDBStateBackend("hdfs://wetech-hdfs:8020/flink/RocksDBBackend"));
            //
            //            CheckpointConfig ckConf = env.getCheckpointConfig();
            //            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            //            ckConf.setCheckpointInterval(10000);
            //            ckConf.setCheckpointTimeout(60000);
            //            ckConf.setMaxConcurrentCheckpoints(1);
            //            ckConf.setMinPauseBetweenCheckpoints(500);
            //
            // ckConf.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            Properties prop = new Properties();
            prop.setProperty("bootstrap.servers", KFKADD);
            prop.setProperty("group.id", "flkCsm1");
            prop.setProperty("auto.offset.reset", "earliest");
            prop.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            prop.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            FlinkKafkaConsumer<String> stringFlinkKafkaConsumer =
                    new FlinkKafkaConsumer<String>(KFKCONTPC, new SimpleStringSchema(), prop);
            DataStreamSource<String> input = env.addSource(stringFlinkKafkaConsumer);

            //            input.print();

            env.execute("Flink-Kafka-Consumer-Test");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
